…; stop deleting shared stream (#385)

## What was broken

When you open a task in the UI, the backend keeps a live stream open to
push updates. That stream had **no way to end itself** — it only stopped
if the browser disconnected.

So when a task finished, the backend just kept listening forever,
holding one Redis connection each time. These pile up until the pool is
empty, at which point the pod can't serve requests (and its health check
fails, so it sits stuck until a manual restart).

A second bug: when one viewer left, the code deleted the task's
**shared** event stream — yanking it out from under everyone else
watching the same task.

## The fix

- **End the stream when the task is done.** The backend already gets a
`task_updated` event when a task finishes, so the stream now closes as
soon as it sees a terminal status.
- **Fallback for the rare cases** the event can't cover (viewer connects
after the task already finished, or the producer dies silently): a
lightweight check on the existing keepalive interval closes the stream
then too.
- **Stop deleting the shared stream** on one viewer's exit — the Redis
TTL already cleans it up on its own.

## Testing

Two integration regression tests (stream self-ends on terminal task; one
viewer disconnecting doesn't delete the shared stream).

**Verified live (A/B against `main`)** — same steps both times: create a
RUNNING task → open `GET /tasks/{id}/stream` → complete the task, while
watching Redis `blocked_clients`.

| | on task completion | Redis connection |
|---|---|---|
| **`main`** | stream keeps `XREAD`-looping; kept sending `:ping` 15s+
after completion and only ended when the client was force-disconnected |
`blocked_clients` stayed `1` — released only on client disconnect
(**zombie**) |
| **this branch** | stream delivers the terminal `task_updated` and
returns immediately (`Ending SSE stream …: received a terminal
task_updated event`) | `blocked_clients` `1 → 0` instantly, no client
disconnect needed |

On `main` the reader logged `Reading messages from Redis stream …` every
~2s indefinitely after the task was done; on this branch it stops the
moment the terminal event arrives. Same reproduction the incident
described, and the connection is released instead of pinned.

> Note: the unit/integration suite runs in CI; the tutorial-agent
integration jobs were red due to a pre-existing missing-`pytest-asyncio`
issue unrelated to this change (fixed separately in
scale-agentex-python).

Related: the UI-side half of this leak is #383.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

<!-- greptile_comment -->

<h3>Greptile Summary</h3>

This PR updates task SSE subscription lifecycle handling.
- Ends subscriptions when a terminal update is received, authoritative
task state becomes terminal, or the task disappears.
- Replays buffered events and closes when connecting to an
already-terminal task.
- Publishes failure updates and centralizes terminal/non-terminal status
classification.
- Preserves shared Redis streams when individual subscribers disconnect.
- Adds integration coverage for terminal shutdown, late connections,
reclaimed keys, transient lookups, and shared-stream retention.

<details><summary><h3>Confidence Score: 4/5</h3></summary>

The PR should not merge until recoverable FAILED tasks no longer
terminate subscriptions that must observe their subsequent RUNNING
updates.

The stream returns upon receiving any FAILED task update, while the task
service intentionally supports forwarding that same FAILED task again
and publishing a RUNNING recovery update; viewers whose subscriptions
closed on FAILED cannot receive that recovery or subsequent events.

**Files Needing Attention:** agentex/src/domain/entities/tasks.py,
agentex/src/domain/use_cases/streams_use_case.py,
agentex/src/domain/services/task_service.py
</details>

<details><summary><h3>Important Files Changed</h3></summary>

| Filename | Overview |
|----------|----------|
| agentex/src/domain/entities/tasks.py | Introduces canonical terminal
and non-terminal status sets shared by state transitions and SSE
termination. |
| agentex/src/domain/services/task_service.py | Publishes FAILED updates
through the normal update path, while existing retry behavior can still
restore FAILED tasks to RUNNING. |
| agentex/src/domain/use_cases/streams_use_case.py | Adds event-driven
and authoritative fallback termination and stops deleting the shared
Redis topic, but the previously reported recoverable-FAILED lifecycle
conflict remains. |
| agentex/src/domain/use_cases/tasks_use_case.py | Reuses the canonical
non-terminal status set for transitions to terminal states. |
| agentex/tests/integration/test_task_stream.py | Adds regression
coverage for terminal shutdown, late subscribers, vanished tasks,
transient lookups, reclaimed keys, and shared-stream preservation. |

</details>

<details><summary><h3>Sequence Diagram</h3></summary>

```mermaid
sequenceDiagram
    participant Client
    participant SSE as StreamsUseCase
    participant DB as Task Store
    participant Redis
    Client->>SSE: Subscribe to task stream
    SSE->>Redis: Snapshot stream cursor
    SSE->>DB: Read authoritative task state
    alt Task already terminal
        SSE->>Redis: Replay buffered events
        SSE-->>Client: Events, then close
    else Task is non-terminal
        loop Until terminal, missing, or disconnect
            SSE->>Redis: Read new events
            alt Terminal task_updated received
                SSE-->>Client: Terminal event
                SSE-->>Client: Close stream
            else Status-check interval elapsed
                SSE->>DB: Recheck task
                alt Terminal or missing
                    SSE-->>Client: Close stream
                end
            end
        end
    end
    Note over SSE,Redis: Subscriber exit does not delete the shared stream
```
</details>

<sub>Reviews (14): Last reviewed commit: ["fix(agentex): terminate SSE
task
streams..."](85dfe6b)
| [Re-trigger
Greptile](https://app.greptile.com/api/retrigger?id=48824816)</sub>

<!-- /greptile_comment -->

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>